package com.zyh.flink.day03;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class ListStateTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> hadoop10 = environment.socketTextStream("hadoop10", 9999);
        KeyedStream<Tuple3<String, String, String>, String> keyedStream = hadoop10.map(new MapFunction<String, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> map(String s) throws Exception {
                String[] strs = s.split("\\s+");
                return Tuple3.of(strs[0],strs[1],strs[2]);
            }
        }).keyBy(t -> t.f0);

        SingleOutputStreamOperator<Tuple2<String, String>> listStateStream = keyedStream.map(new MyListStateRichMapFunction());
        listStateStream.print();

        environment.execute("ListStateJob");
    }
}
class MyListStateRichMapFunction extends RichMapFunction<Tuple3<String,String,String>, Tuple2<String,String>>{
    private ListState<String> listState;

    @Override
    public void open(Configuration parameters) throws Exception {
        //1 获取到RuntimeContext对象
        RuntimeContext ctx = getRuntimeContext();
        //2 创建状态描述符: 状态名和状态中存储的元素类型
        ListStateDescriptor<String> lsd = new ListStateDescriptor<>("categories", Types.STRING);
        //3 根据状态描述符创建状态
        this.listState = ctx.getListState(lsd);
    }

    /*
    * value表示一组上的一个元素 用户一次访问的记录
    * */
    @Override
    public Tuple2<String, String> map(Tuple3<String, String, String> value) throws Exception {
        String newCategory = value.f1;
        //使用状态:把用于最新浏览的商品类别添加到状态中
        //先获取listState中的历史类型,判断是否已经存在
        Iterable<String> oldCategories = this.listState.get();
        Boolean exists = false;
        for (String category : oldCategories) {
            if (category.equals(newCategory)){
                exists = true;
                break;
            }
        }

        if (!exists){
            //不存在则添加新类别
            this.listState.add(newCategory);
        }

        StringBuilder stringBuilder = new StringBuilder();
        for (String category : this.listState.get()) {
            stringBuilder.append(category).append(" ");
        }

        return Tuple2.of(value.f0, stringBuilder.toString());
    }
}